package com.lagou.filter;

import com.lagou.monitor.MonitorWaterLineCalculator;
import org.apache.dubbo.common.constants.CommonConstants;
import org.apache.dubbo.common.extension.Activate;
import org.apache.dubbo.common.utils.CollectionUtils;
import org.apache.dubbo.common.utils.Holder;
import org.apache.dubbo.rpc.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.*;

/**
 * @author xsl
 * @date 2021/5/21 17:49
 * @Description: TPMonitorFilter
 * 调用时长的过滤器添加到消费端
 */
@Activate(group = {CommonConstants.CONSUMER})
public class TPMonitorFilter implements Filter {
    private Logger logger = LoggerFactory.getLogger(TPMonitorFilter.class);
    public final static ConcurrentMap<String, Holder<Object>> monitorMap = new ConcurrentHashMap<>();
    private final static Executor executor = new ThreadPoolExecutor(10, 20, 60
            , TimeUnit.SECONDS, new LinkedBlockingQueue<>(100), Executors.defaultThreadFactory(), new ThreadPoolExecutor.AbortPolicy());
    private static ScheduledExecutorService scheduledExecutor = Executors.newSingleThreadScheduledExecutor();

    public TPMonitorFilter() {
        //每隔5s打印一次最近1分钟内每个方法的TP90、TP99的耗时情况
        scheduledExecutor.scheduleAtFixedRate(() -> {
            if (CollectionUtils.isEmptyMap(monitorMap)) {
                return;
            }
            final Set<Map.Entry<String, Holder<Object>>> entries = monitorMap.entrySet();
            for (Map.Entry<String, Holder<Object>> entry : entries) {
                StringBuilder sb = new StringBuilder("method name:" + entry.getKey());
                List<MonitorWaterLineCalculator> monitors = (List<MonitorWaterLineCalculator>) entry.getValue().get();
                if(CollectionUtils.isNotEmpty(monitors)){
                    for(MonitorWaterLineCalculator monitor:monitors){
                        sb.append(";");
                        sb.append("percent：" + monitor.getPercent());
                        sb.append(";");
                        sb.append("result:" + monitor.getResult());
                    }
                   logger.error(sb.toString());
                }
            }
        }, 10, 5, TimeUnit.SECONDS);
    }

    @Override
    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
        long startTime = System.currentTimeMillis();
        try {
            return invoker.invoke(invocation);
        } finally {
            executor.execute(() -> {
                long elapsedTime = System.currentTimeMillis() - startTime;
//                System.out.println("method name :" + invocation.getMethodName() + ";invoke time :" + elapsedTime);
                List<MonitorWaterLineCalculator> monitors = getMonitors(invocation.getMethodName());
                for (MonitorWaterLineCalculator monitor : monitors) {
                    monitor.accumulate(elapsedTime);
                }
            });
        }
    }

    private List<MonitorWaterLineCalculator> getMonitors(String methodName) {
        Holder<Object> holder = monitorMap.get(methodName);
        if (holder == null) {
            monitorMap.putIfAbsent(methodName, new Holder<>());
        }
        holder = monitorMap.get(methodName);
        Object monitors = holder.get();
        if (monitors == null) {
            synchronized (holder) {
                monitors = holder.get();
                if (monitors == null) {
                    List<MonitorWaterLineCalculator> list = new ArrayList<>(2);
                    list.add(new MonitorWaterLineCalculator(90));
                    list.add(new MonitorWaterLineCalculator(99));
                    holder.set(list);
                }
            }
        }
        return (List<MonitorWaterLineCalculator>) holder.get();
    }


}
